package com.xxd.dfs.namenode;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author: XiaoDong.Xie
 * @create: 2020-08-24 10:59
 * @description: 管理datanode的组件
 */
public class DataNodeManager {
    private final static Logger logger = LoggerFactory.getLogger(DataNodeManager.class);

    private Map<String, DataNodeInfo> datanodes = new ConcurrentHashMap<>();

    private FSNameSystem nameSystem;

    public DataNodeManager(FSNameSystem nameSystem) {
        this.nameSystem = nameSystem;
        new DataNodeAliveMonitor().start();
    }

    /**
     * 注册请求
     *
     * @param dataNodeInfo
     */
    public boolean register(DataNodeInfo dataNodeInfo) {
        String key = dataNodeInfo.getId();
        if (datanodes.containsKey(key)) {
            logger.warn("dataNodeInfo = {} 已经注册过了，注册失败", dataNodeInfo);
            return false;
        }

        datanodes.put(key, dataNodeInfo);
        logger.info("{} 注册成功", dataNodeInfo);
        return true;
    }

    /**
     * @param ip
     * @param hostname
     * @return
     */
    public boolean heartbeat(String ip, String hostname) {
        String key = getDataNodeId(ip, hostname);
        DataNodeInfo dataNodeInfo = datanodes.get(key);

        if (dataNodeInfo == null) {
            // 这个时候就需要指示DataNode 重新注册以及全量上报
            logger.warn(" ip = {} hostname = {} 心跳失败，需要重新注册.......", ip, hostname);
            return false;
        }

        dataNodeInfo.setLatestHeartbeatTime(System.currentTimeMillis());
        logger.info("{} 心跳接收成功", dataNodeInfo);
        return true;
    }

    public String getDataNodeId(String ip, String hostname) {
        return ip + "_" + hostname;
    }

    public Map<String, DataNodeInfo> getDatanodes() {
        return datanodes;
    }

    /**
     * 分配datanode
     * hadoop 中使用机架感知 + 负载均衡算法 分配datanode
     * 我们这里没有机架感知的业务，直接保证一个负载均衡就行了
     *
     * @param fileSize
     * @return
     */
    public List<DataNodeInfo> allocateDataNodes(long fileSize) {
        // 选出容量最少的两台datanode
        // 这里就涉及到算法了
        // 我们可以用最挫的 排序 来解决，可以使用排序容器吗？
        // 如果用的话 并发安全怎么考虑
        List<DataNodeInfo> datanodeList = new ArrayList<>(datanodes.values());

        Collections.sort(datanodeList);

        // 选择存储数据最少的头两个datanode出来
        List<DataNodeInfo> selectedDatanodes = new ArrayList<DataNodeInfo>(2);
        if (datanodeList.size() >= 2) {
            selectedDatanodes.add(datanodeList.get(0));
            selectedDatanodes.add(datanodeList.get(1));

            // 这里需要直接增加datanode 的容量吗，需要的，这样容易做负载均衡
            datanodeList.get(0).addStoredDataSize(fileSize);
            datanodeList.get(1).addStoredDataSize(fileSize);
        }
        return selectedDatanodes;
    }

    /**
     * 心跳检测
     * 会循环检测，30s检查一次
     * 10分钟没有心跳就认为宕机
     */
    private class DataNodeAliveMonitor extends Thread {
        @Override
        public void run() {
            while (nameSystem.isRunning()) {
                List<DataNodeInfo> toRemoveDatanodes = new ArrayList<>();
                for (Map.Entry<String, DataNodeInfo> nodeInfoEntry : datanodes.entrySet()) {
                    DataNodeInfo dataNodeInfo = datanodes.get(nodeInfoEntry.getKey());
                    if (System.currentTimeMillis() - dataNodeInfo.getLatestHeartbeatTime() > 10 * 60 * 1000) {
                        toRemoveDatanodes.add(dataNodeInfo);
                    }
                }

                if (!toRemoveDatanodes.isEmpty()) {
                    for (DataNodeInfo toRemoveDatanode : toRemoveDatanodes) {
                        logger.warn("数据节点【{}】宕机，需要 进行副本复制......", toRemoveDatanode);

                        // 还有一个地方是在哪儿呢？就是这个叫做，我们还维护了一块是数据节点和副本的关系
                        // createLostReplicaTask(toRemoveDatanode);

                        // 来实现DataNode宕机之后的一些处理
                        logger.info("正在移除 【{}】节点，在datanodeManager 中的信息", toRemoveDatanode);
                        datanodes.remove(toRemoveDatanode.getId());

                        // 删除掉这个数据结构
                        // namesystem.removeDeadDatanode(toRemoveDatanode);
                    }
                }

                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

